package com.yufei.infoExtractor.parallel.executor;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.yufei.infoExtractor.parallel.ParallelThreadExecutor;
import com.yufei.infoExtractor.parallel.thread.UrlParserThread;
import com.yufei.infoExtractor.util.CommonUtil;
import com.yufei.infoExtractor.util.ExceptionUtil;

public class UrlExtractorParallelThreadExecutor implements
		ParallelThreadExecutor {
	public static final Log mLog = LogFactory.getLog(UrlExtractorParallelThreadExecutor.class);

	@Override
	public void parallelExecuteThreadsForCollectionData(Collection ks,
			int blockSize, Map context) {
		// TODO Auto-generated method stub
		List<List<String>> urlsGroup=CommonUtil.splitKs(ks, blockSize);
		int groupCount=urlsGroup.size();
		CountDownLatch countDownLatch = new CountDownLatch(groupCount);
		//主线程实例化一个线程安全的threadLocal变量，这样子线程就可以安全的方位某些共享对象
		ThreadLocal<Map> threadLocalcontext=new ThreadLocal<Map>();
		threadLocalcontext.set(context);
		Runnable executorThread=null;
			ExecutorService executorService=Executors.newFixedThreadPool(groupCount);
			int threadNumber=0;
			List<String> urlList=null;
			for(int i=0;i<urlsGroup.size();i++){
				urlList=urlsGroup.get(i);
				threadNumber+=1;
				
				executorThread =new UrlParserThread(threadLocalcontext,urlList, countDownLatch);
			    executorService.execute(executorThread);

			}
			try {
				countDownLatch.await();
				
	

			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
			mLog.info(ExceptionUtil.getExceptionDetailsMessage(e));
			}finally{
				executorService.shutdown();
			}
	}



}
